package com.wq.hadoop.hbase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author: wangqiang20995
 * @Date:2018/11/4
 * @Description:
 * @Resource:
 */
public class HBaseClientDemo {

    private static final String HBASE_DIR = "hbase.rootdir";
    private static final String ZOOKEEPER_NODE = "hbase.zookeeper.quorum";

    private static Logger logger = Logger.getLogger(HBaseClientDemo.class);

    private static Configuration configuration;

    private static Connection connection;

    static {
        configuration = HBaseConfiguration.create();
        configuration.set(HBASE_DIR,"hdfs://192.168.56.3:9000/hbase");
        configuration.set(ZOOKEEPER_NODE,"192.168.56.3");

        try {
            logger.info(System.getenv("HADOOP_HOME"));
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            logger.error(e.getMessage(),e);
        }
    }


    public static void main(String args[]) throws IOException {

        Admin admin = connection.getAdmin();

//        existTable(admin);

//        createTable(admin,"people",new String[]{"basic","money"});

//        listTables(admin);

//        modify("people","001","basic","name","George");

//        information("people","001","basic","address");

        scanTable("people");
        doFinal(admin);

    }


    private static void scanTable(String tableName) throws IOException {
        HTable hTable = new HTable(configuration,tableName);

        Scan scan = new Scan();
        scan.addColumn("basic".getBytes(),"name".getBytes());
        scan.addColumn("basic".getBytes(),"address".getBytes());
        scan.addColumn("money".getBytes(),"receive".getBytes());


        ResultScanner results = hTable.getScanner(scan);
        for(Result result = results.next(); result != null;result = results.next()){
            logger.info(String.format("found row --> [%s]",result));
        }

        if(results != null){
            results.close();
        }

        if(hTable != null){
            hTable.close();
        }
    }

    private static void information(String table,String rowId,String family,String column) throws IOException {
        HTable hTable = new HTable(configuration,table);
        Get get = new Get(Bytes.toBytes(rowId));
        Result result = hTable.get(get);

        byte[] values = result.getValue(Bytes.toBytes(family),Bytes.toBytes(column));
        logger.info(String.format("table[%s]--rowId[%s]--family[%s]--column[%s]--->value[%s]",table,rowId,family,column,new String(values)));

        if(hTable != null){
            hTable.close();
        }
    }

    private static void modify(String table,String rowId,String columnFamily,String column,String value) throws IOException {
        HTable hTable = new HTable(configuration,table);
        Put put = new Put(Bytes.toBytes(rowId));
        put.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(column),Bytes.toBytes(value));
        hTable.put(put);
        logger.info("put or modify column done");
        if(hTable != null){
            hTable.close();
        }

    }

    private static void listTables(Admin admin) throws IOException {
        HTableDescriptor[] tableDescriptores = admin.listTables();
        if(tableDescriptores == null){
            logger.error("HBase中表描述对象返回对象为空");
        }

        List<Map<String,Object>> metaTables = new ArrayList<>(tableDescriptores.length);
        StringBuilder builder = new StringBuilder();
        for(HTableDescriptor descriptor : tableDescriptores){
            Map<String,Object> metaTable = new HashMap<>();
            String name = descriptor.getNameAsString();
            metaTable.put("TableName",name);

            HColumnDescriptor[] columns = descriptor.getColumnFamilies();
            List<Map<String,Object>> list = new ArrayList<>();
            for(HColumnDescriptor columnDescriptor : columns){
                Map<String,Object> map = new HashMap<>();
                String column = columnDescriptor.getNameAsString();
                Map<String,String> columnIdentities = columnDescriptor.getConfiguration();
                map.put("ColumnFamily",column);
                map.put("Columns",columnIdentities);
                list.add(map);
            }

            metaTable.put("ColumnFamily",list);
            metaTables.add(metaTable);
        }

        logger.info(String.format("list table meta from HBase [%s]",metaTables));
    }

    private static void createTable(Admin admin,String tableName,String... columnFamilies) throws IOException {
        HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
        if(columnFamilies == null){
            logger.error("列族不能为空");
        }
        for(String columnFamily : columnFamilies){
            HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnFamily);
            tableDescriptor.addFamily(columnDescriptor);
        }

        admin.createTable(tableDescriptor);
        logger.info(String.format("create table in hbase with name[%s] and column families[%s]",tableName,columnFamilies));
    }

    private static void existTable(Admin admin) throws IOException {
        TableName tableName = TableName.valueOf("account");
        logger.info("account exist or not --> " + admin.tableExists(tableName));
    }

    private static void doFinal(Admin admin) throws IOException {
        if(admin != null){
            admin.close();
        }

        if(connection != null){
            connection.close();
        }
    }
}
